//package com.generator.controller.impl;
//
//import com.alibaba.dubbo.common.utils.StringUtils;
//import com.generator.config.ServiceConfig;
//import com.generator.constant.NodePath;
//import com.generator.controller.LeaderService;
//import com.generator.mode.HostAndPort;
//import com.generator.mode.LeaderInfo;
//import com.generator.mode.NodeInfo;
//import com.generator.service.ZkService;
//import com.generator.utils.JsonUtils;
//import com.generator.utils.NetUtils;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.zookeeper.CreateMode;
//import org.springframework.beans.factory.InitializingBean;
//import org.springframework.context.annotation.Lazy;
//import org.springframework.core.PriorityOrdered;
//import org.springframework.core.annotation.Order;
//import org.springframework.stereotype.Service;
//
//import javax.annotation.Resource;
//import java.util.concurrent.ThreadLocalRandom;
//import java.util.concurrent.TimeUnit;
//import java.util.concurrent.atomic.AtomicInteger;
//import java.util.concurrent.locks.LockSupport;
//
//@Order(PriorityOrdered.HIGHEST_PRECEDENCE)
//@Lazy(false)
//@Slf4j
//@Service
//public class LeaderServiceImpl implements LeaderService, InitializingBean {
//
//    @Resource
//    private ServiceConfig serviceConfig;
//
//    @Resource
//    private ZkService zkService;
//
//    private final AtomicInteger epoch = new AtomicInteger(0);
//
//    /**
//     * 状态: -1: 选举中; 0: 初始状态; 1:选举结束
//     */
//    private final AtomicInteger status = new AtomicInteger(0);
//
//    private HostAndPort hostAndPort;
//
//    private volatile boolean isLeader = false;
//
//    @Override
//    public void afterPropertiesSet() {
//
//        hostAndPort = HostAndPort.builder().host(NetUtils.getHostForIPV4()).port(serviceConfig.getPort()).build();
//
//        electLeader();
//
//        joinCluster();
//    }
//
//    @Override
//    public LeaderInfo electLeader() {
//
//        log.info("[LeaderService] 集群Leader选举启动 .....");
//
//        // 选举中
//        status.set(-1);
//
//        LeaderInfo leaderInfo = new LeaderInfo();
//        leaderInfo.setHostAndPort(hostAndPort);
//
//        log.info("[LeaderService] 使用 {} 值创建节点【{}】并更新其数据.",
//                leaderInfo,
//                NodePath.CONTROLLER_PATH);
//
//        boolean success = zkService.createPath(NodePath.CONTROLLER_PATH,JsonUtils.toJson(leaderInfo), CreateMode.EPHEMERAL);
//
//        if(success) {
//
//            log.info("[LeaderService] 当前节点成为集群Leader",NodePath.CONTROLLER_PATH,hostAndPort);
//
//            this.isLeader = true;
//
//        } else {
//
//            String data = zkService.getData(NodePath.CONTROLLER_PATH);
//
//            leaderInfo = JsonUtils.fromJson(data,LeaderInfo.class);
//
//            log.info("[LeaderService] The node {} exists,it's value is {}",NodePath.CONTROLLER_PATH,data);
//            this.isLeader = leaderInfo.getHostAndPort().equals(hostAndPort);
//        }
//
//        log.info("[LeaderService] 集群Leader选举结束, 集群Leader为: host:{},port:{}",
//                leaderInfo.getHostAndPort().getHost(),
//                leaderInfo.getHostAndPort().getPort());
//
//        afterElectLeader();
//
//        status.compareAndSet(-1,1);
//
//        return leaderInfo;
//    }
//
//    @Override
//    public Integer incrementEpoch() {
//        checkAndWaitElection();
//        zkService.setData(NodePath.EPOCH_PATH,(epoch.get()+1)+"");
//        return epoch.incrementAndGet();
//    }
//
//    @Override
//    public Integer getEpoch() {
//        checkAndWaitElection();
//        return epoch.get();
//    }
//
//    @Override
//    public boolean isLeader() {
//        checkAndWaitElection();
//        return isLeader;
//    }
//
//    @Override
//    public LeaderInfo getLeader() {
//        checkAndWaitElection();
//        return getLeaderFromRemote();
//    }
//
//    @Override
//    public void checkAndWaitElection() {
//        while (!Integer.valueOf(1).equals(status.get())) {
//            Thread.yield();
//        }
//    }
//
//    private Integer getRemoteEpoch() {
//        String value = zkService.getData(NodePath.EPOCH_PATH);
//        return StringUtils.isBlank(value) ? epoch.get() : Integer.valueOf(value);
//    }
//
//    /**
//     * 注册节点信息
//     */
//    private void joinCluster() {
//        // 注册节点信息
//        NodeInfo nodeInfo = new NodeInfo();
//        nodeInfo.setHostAndPort(hostAndPort);
//        nodeInfo.setEpoch(this.epoch.get());
//        nodeInfo.setEndOffset(0);
//        zkService.createPath(NodePath.CLUSTER_PATH + NodePath.SEPARATOR + NetUtils.getHostForIPV4()+":"+serviceConfig.getPort(),JsonUtils.toJson(nodeInfo),CreateMode.EPHEMERAL);
//    }
//
//    private void afterElectLeader() {
//        resolveEpoch();
//    }
//
//    private void resolveEpoch() {
//
//        if(isLeader) {
//
//            log.info("[LeaderService] 当前节点为Leader, 将准备升级epoch值");
//
//            // 只有leader节点可以更新epoch
//
//            boolean epochNodeExists = zkService.exists(NodePath.EPOCH_PATH);
//
//            if(epochNodeExists) {
//
//                Integer remoteEpoch = getRemoteEpoch();
//                epoch.set(remoteEpoch+1);
//                zkService.setData(NodePath.EPOCH_PATH,String.valueOf(epoch.get()));
//
//            } else {
//
//                // epoch 节点不存在
//                log.info("[LeaderService] 创建 {} 节点",NodePath.EPOCH_PATH);
//                zkService.createPath(NodePath.EPOCH_PATH,String.valueOf(epoch.incrementAndGet()),CreateMode.PERSISTENT);
//            }
//
//            log.info("[LeaderService] 当前节点为Leader, 升级后的epoch值:{}",epoch.get());
//
//        } else {
//
//            log.info("[LeaderService] 当前节点为Follower, 短暂等待500毫秒,等待Leader节点更新epoch");
//
//            // 随机等待一段时间,等待Leader更新epoch
//            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(ThreadLocalRandom.current().nextInt(500)));
//
//            Integer remoteEpoch = getRemoteEpoch();
//
//            log.info("[LeaderService] 当前节点为Follower, epoch最新值为:{}",remoteEpoch);
//
//            // 更新本地epoch值
//            epoch.set(remoteEpoch);
//        }
//    }
//
//    private LeaderInfo getLeaderFromRemote() {
//        String value = zkService.getData(NodePath.CONTROLLER_PATH);
//        return JsonUtils.fromJson(value,LeaderInfo.class);
//    }
//
//}
